/*
* This file is a part of the open-eBackup project.
* This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
* If a copy of the MPL was not distributed with this file, You can obtain one at
* http://mozilla.org/MPL/2.0/.
*
* Copyright (c) [2024] Huawei Technologies Co.,Ltd.
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
*/
#include "ReplicationCommonService.h"
#include "client/ClientInvoke.h"
#include "ReplicationDeviceManager.h"
#include "utils/PluginUtilities.h"
#include "system/System.hpp"
#include "utils/CertMgr.h"
#include "common/Utils.h"

using namespace AppProtect;
using namespace nasreplicationservice;
using namespace Module;
using namespace NasChannelManager;

namespace {
    constexpr auto MODULE = "ReplicationBackupJob";
    constexpr uint16_t SCAN_TIME_INTERVAL_SECEND = 5;
    constexpr int SCAN_RETRY_TIMES = 720; // 每隔5s扫描一次，超时时间1小时
    constexpr int RETRY_TIMES = 3;
    constexpr int SNAP_CONVERT = 2;
    constexpr uint16_t SYNCHRONIZE_FINISHED = 97;
    constexpr float PROGRESS_OVERALL = 100.0;
    constexpr int32_t PROGRESS_HUNDRED = 100;
    constexpr int32_t PROGRESS_TEN = 10;
    constexpr int32_t PROGRESS_TWENTY = 20;
    constexpr int32_t SPEED_INIT_VALUE = 100;
    constexpr int MAX_ROLLBACK_RATE = 100;
    constexpr auto RETAINED_LOGIN_TIMES = "1";
    constexpr int EXECEPTION_RETRY_TIME = 6;
    constexpr int ADD_NEW_JOB_RETRY_INTERVAL = 60;
    constexpr uint32_t REPORT_RUNNING_INTERVAL = 10;
    constexpr uint32_t REPORT_RUNNING_TIMES = 12;
    constexpr uint64_t QUERY_FILESYSTEM_SLEEP_TIME = 40; // 查询文件系统容量，需要延迟，否则查询不准
}

EXTER_ATTACK int ReplicationCommonService::AllowBackupInLocalNode()
{
    return MP_SUCCESS;
}

EXTER_ATTACK int ReplicationCommonService::CheckBackupJobType()
{
    return MP_SUCCESS;
}

int ReplicationCommonService::PrerequisiteJobInner()
{
    HCP_Log(INFO, MODULE) << "Begin to exeute pre job: job id " << m_parentJobId << HCPENDLOG;
    SubJobDetails subJobDetails;
    LogDetail logDetail{};
    ActionResult result;
    std::vector<LogDetail> logDetailList;
    ABORT_ENDTASK(subJobDetails, result, logDetailList, logDetail, 0, 0);
    AddLogDetail(logDetail, "", JobLogLevel::TASK_LOG_INFO);
    REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, 0, 0, SubJobStatus::RUNNING);
    // 监控线程
    HCP_Log(INFO, MODULE) << "Start new thread for report : " << m_parentJobId << HCPENDLOG;
    m_connecting = true;
    std::thread monitorJobThread(&ReplicationCommonService::MonitorJobThread, this, REPORTSLEEPTIME);
    if (!InitializeTask()) {
        HCP_Log(ERR, MODULE) << "reporting pre job failed, job id:" << m_parentJobId << HCPENDLOG;
        m_connecting = false;
        monitorJobThread.join();
        ABORT_ENDTASK(subJobDetails, result, logDetailList, logDetail, 0, 0);
        return MP_FAILED;
    }
    if (!AfterInitializeTask()) {
        HCP_Log(ERR, MODULE) << "reporting pre job failed, job id:" << m_parentJobId << HCPENDLOG;
        m_connecting = false;
        monitorJobThread.join();
        ABORT_ENDTASK(subJobDetails, result, logDetailList, logDetail, 0, 0);
        return MP_FAILED;
    }
    m_connecting = false;
    monitorJobThread.join();

    ABORT_ENDTASK(subJobDetails, result, logDetailList, logDetail, 0, 0);
    REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, 0, 0, SubJobStatus::COMPLETED);
    HCP_Log(INFO, MODULE) << "Finish to exeute pre job: job id" << m_parentJobId << HCPENDLOG;
    return MP_SUCCESS;
}

EXTER_ATTACK int ReplicationCommonService::PrerequisiteJob()
{
    int ret = PrerequisiteJobInner();
    SetJobToFinish();
    return ret;
}

int ReplicationCommonService::GenerateSubJobInner()
{
    HCP_Log(INFO, MODULE) << "Generate sub task, task id is " << m_parentJobId << HCPENDLOG;
    SubJob subJob;
    subJob.__set_jobId(m_parentJobId);
    subJob.__set_jobType(SubJobType::BUSINESS_SUB_JOB);
    subJob.__set_jobName("RemoteReplication");
    subJob.__set_policy(ExecutePolicy::ANY_NODE); // LOCAL_NODE
    std::vector<SubJob> vec;
    vec.push_back(subJob);

    SubJobDetails subJobDetails;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetailList;
    ActionResult result;
    ABORT_ENDTASK(subJobDetails, result, logDetailList, logDetail, 0, 0);

    int retryCnt = 0;
    while (true) {
        JobService::AddNewJob(result, vec);
        if (result.code == 0) {
            break;
        }
        ++retryCnt;
        if (retryCnt >= EXECEPTION_RETRY_TIME) {
            AddLogDetail(logDetail, "plugin_generate_subjob_fail_label", JobLogLevel::TASK_LOG_ERROR);
            REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, 0, 0, SubJobStatus::FAILED);
            return MP_FAILED;
        }
        sleep(ADD_NEW_JOB_RETRY_INTERVAL);
    }
    REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, 0, 0, SubJobStatus::COMPLETED);
    HCP_Log(INFO, MODULE) << "Finish to generate sub job, main task id is " << m_parentJobId << HCPENDLOG;
    return MP_SUCCESS;
}

EXTER_ATTACK int ReplicationCommonService::GenerateSubJob()
{
    HCP_Log(INFO, MODULE) << "Enter ReplicationCommonService GenerateSubJob()" << HCPENDLOG;
    m_generateSubjobFinish = false;
    std::thread keepAlive = std::thread(&ReplicationCommonService::KeepPluginAlive, this);
    int ret = GenerateSubJobInner();
    SetJobToFinish();
    m_generateSubjobFinish = true;
    if (keepAlive.joinable()) {
        keepAlive.join();
        INFOLOG("keep alive thread join!");
    }
    return ret;
}

void ReplicationCommonService::KeepPluginAlive()
{
    HCP_Log(INFO, MODULE) << "Enter KeepPluginAlive" << HCPENDLOG;
    ActionResult result;
    SubJobDetails subJobDtls;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetails;
    m_curProcess = PROGRESS_TEN;
    uint32_t reportCnt = 0;
    while (!m_isAbort && !m_generateSubjobFinish) {
        // 10s 检查一次退出条件
        std::this_thread::sleep_for(std::chrono::seconds(REPORT_RUNNING_INTERVAL));
        // 120s 上报一次
        if (reportCnt % REPORT_RUNNING_TIMES == 0) {
            REPORT_LOG2AGENT(subJobDtls, result, logDetails, logDetail, m_curProcess, 0, SubJobStatus::RUNNING);
        }
        reportCnt++;
    }
}
int ReplicationCommonService::ExecuteSubJobInner()
{
    std::string subJobId = m_subJobInfo->subJobId;
    HCP_Log(INFO, MODULE) << "ExecuteSubJob: main job id:" << m_parentJobId << ", subJob id:" << subJobId << HCPENDLOG;
    ActionResult result;
    SubJobDetails subJobDtls;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetails;
    m_curProcess = PROGRESS_TEN;
    REPORT_LOG2AGENT(subJobDtls, result, logDetails, logDetail, m_curProcess, 0, SubJobStatus::RUNNING);
    // 监控线程
    HCP_Log(INFO, MODULE) << "Start new thread for report : " << m_parentJobId << ", " << subJobId << HCPENDLOG;
    std::thread monitorJobThread(&ReplicationCommonService::MonitorJobThread, this, REPORTSLEEPTIME);
    m_connecting = true;
    if (!StartTask(m_localDeviceController)) {
        HCP_Log(ERR, MODULE) << "Start ExecuteSubJobInner fail, main job id is " << m_parentJobId << HCPENDLOG;
        m_connecting = false;
        monitorJobThread.join();
        return MP_FAILED;
    }
    m_connecting = false;
    monitorJobThread.join();
    m_curProcess = PROGRESS_TWENTY;
    REPORT_LOG2AGENT(subJobDtls, result, logDetails, logDetail, m_curProcess, 0, SubJobStatus::RUNNING);
    if (!StartMonitorTask(m_localDeviceController, m_remoteDeviceController)) {
        return MP_FAILED;
    }
    if (IsAbortJob()) {
        HCP_Log(INFO, MODULE) << "MonitorTask is aborted, main job id is " << m_parentJobId << HCPENDLOG;
        REPORT_LOG2AGENT(subJobDtls, result, logDetails, logDetail, m_curProcess, m_avgSpeed, SubJobStatus::ABORTED);
        return MP_FAILED;
    }

    uint64_t endBackupTime = time(nullptr); // SetDataSizeForJobDetail有sleep，会影响平均速度计算，备份结束时间提前记录
    
    SetDataSizeForJobDetail(subJobDtls);
    if (m_isBackUpFlow) {
        AddLogDetail(logDetail, "nas_plugin_backup_filesystem_success_label", JobLogLevel::TASK_LOG_INFO);
        uint64_t totalBackupTime = endBackupTime - m_satrtTransTime;
        if (m_satrtTransTime != 0 && totalBackupTime != 0) {
            m_avgSpeed = m_dataSize / totalBackupTime;
        }
        HCP_Log(INFO, MODULE) << "Backup startTime is " << m_satrtTransTime << ", EndTime is " << endBackupTime
                              << ",totalBackupTime is " << totalBackupTime << ",m_avgSpeed is " << m_avgSpeed
                              << ",m_dataSize is " << m_dataSize << HCPENDLOG;
    } else {
        AddLogDetail(logDetail, "nas_plugin_restore_filesystem_success_label", JobLogLevel::TASK_LOG_INFO);
    }
    REPORT_LOG2AGENT(subJobDtls, result, logDetails, logDetail, PROGRESS_HUNDRED, m_avgSpeed, SubJobStatus::COMPLETED);
    m_isReportData = false;
    HCP_Log(INFO, MODULE) << "Finish to execute sub job. Main job id:" << m_parentJobId
        << ", sub job id:" << subJobId << HCPENDLOG;
    return MP_SUCCESS;
}

bool ReplicationCommonService::StartMonitorTask(std::shared_ptr<DoradoNas>& localController,
    std::shared_ptr<DoradoNas>& remoteController)
{
    ActionResult result;
    SubJobDetails subJobDtls;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetails;
    int64_t errorCode = MonitorTask(m_localDeviceController, m_remoteDeviceController);
    if (errorCode == MP_SUCCESS) {
        return true;
    }
    HCP_Log(ERR, MODULE) << "Execute MonitorTask failed, main job id is " << m_parentJobId << HCPENDLOG;
    if (m_isBackUpFlow) {
        AddLogDetail(logDetail, "nas_plugin_backup_filesystem_fail_label", JobLogLevel::TASK_LOG_ERROR);
    } else {
        AddLogDetail(logDetail, "nas_plugin_restore_filesystem_fail_label", JobLogLevel::TASK_LOG_ERROR);
    }
    if (IsReportAgentErrorCode(logDetail)) {
        return false;
    }
    AddErrCode(logDetail, errorCode);
    REPORT_LOG2AGENT(subJobDtls, result, logDetails, logDetail, m_curProcess, m_avgSpeed, SubJobStatus::FAILED);
    return false;
}

void ReplicationCommonService::SetDataSizeForJobDetail(SubJobDetails &subJobDetails)
{
    /* 备份时m_isReportData设置为ture，恢复为false */
    if (!m_isReportData) {
        return;
    }
    DeviceDetails info;
    HCP_Log(DEBUG, MODULE) << "sleep before query filesystem."  << HCPENDLOG;
    sleep(QUERY_FILESYSTEM_SLEEP_TIME);
    uint64_t preUsedCapacity = 0;
    INFOLOG("capacityPath: %s", m_capacityPath.c_str());
    /* 可靠性: 通过文件记录查询第一次备份任务开始前的文件系统容量，如果没有查询到，则使用当前子任务开始前查询到的容量 */
    std::vector<std::string> fileContent;
    if (!PluginUtils::IsFileExist(m_capacityPath) || !PluginUtils::ReadFile(m_capacityPath, fileContent)) {
        WARNLOG("Read filesytem pre use capacity failed.");
    }
    preUsedCapacity = !fileContent.empty() ? Module::SafeStoUll(fileContent[0], m_usedCapacity) : m_usedCapacity;
    if (m_remoteDeviceController->Query(info) != MP_SUCCESS) {
        WARNLOG("Query remote device info failed.");
    }
    int64_t usedCapacity = (info.usedCapacity > preUsedCapacity) ? info.usedCapacity - preUsedCapacity : 0;
    m_dataSize = usedCapacity / 2;  // 扇区大小是512 bytes, 除以2才是真实的kb单位
    HCP_Log(INFO, MODULE) << "Report data size: " << m_dataSize << "kb" << HCPENDLOG;
    subJobDetails.__set_dataSize(m_dataSize);
}

EXTER_ATTACK int ReplicationCommonService::ExecuteSubJob()
{
    AddSubJobToChannelMgr();
    std::shared_ptr<void> delSubJob(nullptr, [&](...) {
        RemoveSubJobFromChannelMgr();
    });
    int ret = ExecuteSubJobInner();
    SetJobToFinish();
    return ret;
}

EXTER_ATTACK int ReplicationCommonService::PostJob()
{
    // 监控线程
    HCP_Log(INFO, MODULE) << "Start new thread for report post job : " << m_parentJobId << HCPENDLOG;
    AddSubJobToChannelMgr();
    std::shared_ptr<void> delSubJob(nullptr, [&](...) {
        RemoveSubJobFromChannelMgr();
    });
    std::thread monitorJobThread(&ReplicationCommonService::MonitorJobThread, this, REPORTSLEEPTIME);
    m_connecting = true;
    int ret = PostJobInner();
    m_connecting = false;
    monitorJobThread.join();
    DeleteUsedCapacityResource();
    SetJobToFinish();
    return ret;
}

bool ReplicationCommonService::GetMgrIpFromExt(std::vector<std::string>& ipList, const std::string& extInfo)
{
    // extInfo: { "managerIps" : "[\"8.40.97.165\",\"8.40.97.166\"]"}
    Json::Value value;
    if (!Module::JsonHelper::JsonStringToJsonValue(extInfo, value)) {
        ERRLOG("GetMgrIpFromExt JsonStringToJsonValue failed");
        return false;
    }

    Json::Value mgrIpValueList = value["managerIps"];
    std::string mgrIpString = mgrIpValueList.asString();

    Json::Value mgrIpValue;
    if (!Module::JsonHelper::JsonStringToJsonValue(mgrIpString, mgrIpValue)) {
        ERRLOG("GetMgrIpFromExt JsonStringToJsonValue failed");
    }

    for (const auto& ipValue : mgrIpValue) {
        std::string ip = ipValue.asString();
        DBGLOG("GetMgrIpFromExt mgr ip: %s", ip.c_str());
        ipList.push_back(ip);
    }

    if (ipList.empty()) {
        ERRLOG("GetMgrIpFromExt ip list in extInfo is empty");
        return false;
    }
    std::for_each(ipList.begin(), ipList.end(), [&] (const std::string& ip) { INFOLOG("ip: %s", ip.c_str());});

    std::string objList;
    for (std::size_t i = 0; i < ipList.size(); ++i) {
        objList.append(ipList[i]);
        if (i == ipList.size() - 1) {
            break;
        }
        objList.append(",");
    }
    m_productManageIps = objList; // 默认使用多控ip，用于为初始化device之前的网络校验
    INFOLOG("m_productManageIps: %s", m_productManageIps.c_str());
    return true;
}

bool ReplicationCommonService::GetRepIpFromExt(const std::string& extInfo)
{
    // extinfo : {"repIps" : "[{\"NAME\":\"CET_A_P0P2_DATA\",\"IPV4ADDR\":\"192.168.129.170\"},
    // {\"NAME\":\"CET_b_P0P2_DATA\",\"IPV4ADDR\":\"192.168.129.171\"}]"}
    Json::Value value;
    std::string repIpListStr;
    if (!Module::JsonHelper::JsonStringToJsonValue(extInfo, value)) {
        ERRLOG("Convert JsonStringToJsonValue failed");
        return false;
    }

    Json::Value repIpList = value["repIps"];
    try {
        repIpListStr = repIpList.asString();
    } catch (const std::exception& erro) {
        HCP_Log(ERR, MODULE) << "Error occured in repIpList to string: " << erro.what() << HCPENDLOG;
        return false;
    }
    if (repIpListStr.empty()) {
        return true;
    }

    Json::Value repIpValue;
    if (!Module::JsonHelper::JsonStringToJsonValue(repIpListStr, repIpValue)) {
        ERRLOG("Convert JsonStringToJsonValue failed");
        return false;
    }

    for (auto& ipValue : repIpValue) {
        ApplicationEnvRepIp repIp;
        if (!Module::JsonHelper::JsonValueToStruct(ipValue, repIp)) {
            ERRLOG("Convert JsonStringToJsonValue failed");
            return false;
        }
        m_selectedRepNameList.insert(repIp.name);
        INFOLOG("Get Rep Ips: %s, %s", repIp.name.c_str(), repIp.ipv4addr.c_str());
    }
    return true;
}

bool ReplicationCommonService::UpdateDeviceAvilableIP(ControlDeviceInfo &deviceInfo,
    const std::vector<std::string> &mgrIpList)
{
    ActionResult result;
    SubJobDetails subJobDetails;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetailList;
    REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, 0, SubJobStatus::RUNNING);
    bool isFindMrgIp = false;
    for (auto mgrIpPtr = mgrIpList.rbegin(); mgrIpPtr != mgrIpList.rend(); ++mgrIpPtr) {
        deviceInfo.url = *mgrIpPtr;
        std::string esn;
        // 先取本地的， 检查连通性， 没有的话再创建， 再检查连通性
        std::shared_ptr<DoradoNas> deviceControl = ReplicationDeviceManager::GetInstance().GetDeviceByIp(*mgrIpPtr);
        HCP_Log(INFO, MODULE) << "Get device from map: " << *mgrIpPtr << HCPENDLOG;
        if (deviceControl != nullptr) {
            deviceControl->SetRetryAttr(1);
            if (deviceControl->GetESN(esn) == MP_SUCCESS && !esn.empty()) {
                REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail,
                    m_curProcess, 0, SubJobStatus::RUNNING);
                isFindMrgIp = true;
                m_productManageIps = *mgrIpPtr; // 用于故障切换时的ip校验,和deviceControl所使用的ip对应
                break;
            } else {
                ReplicationDeviceManager::GetInstance().RemoveDeviceFromMap(*mgrIpPtr);
            }
        }
        deviceControl = std::make_shared<DoradoNas>(deviceInfo, false);
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, 0, SubJobStatus::RUNNING);
        deviceControl->SetRetryAttr(1, 1); // 设置重试次数为1， 且等待时间为1s
        if (deviceControl->GetESN(esn) == MP_SUCCESS && !esn.empty()) {
            REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, 0, SubJobStatus::RUNNING);
            HCP_Log(INFO, MODULE) << "Add device to map : " << *mgrIpPtr << HCPENDLOG;
            ReplicationDeviceManager::GetInstance().AddDevice(*mgrIpPtr, deviceControl);
            isFindMrgIp = true;
            m_productManageIps = *mgrIpPtr; // 用于故障切换时的ip校验,和deviceControl所使用的ip对应
            break;
        }
    }
    if (!isFindMrgIp) {
        AddLogDetail(logDetail, "nas_plugin_init_env_fail_label", JobLogLevel::TASK_LOG_ERROR);
        if (!PluginUtils::CheckDeviceNetworkConnect(m_productManageIps)) {
            return false;
        }
        AddErrCode(logDetail, HomoErrorCode::LOGIN_FAILED_ERROR_CODE, RETAINED_LOGIN_TIMES);
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, 0, SubJobStatus::FAILED);
        HCP_Log(ERR, MODULE) << "Found no available ip in device." << HCPENDLOG;
        return false;
    }
    HCP_Log(INFO, MODULE) << "Get available device url: " << WIPE_SENSITIVE(deviceInfo.url) << HCPENDLOG;
    return true;
}

bool ReplicationCommonService::AfterInitializeTask()
{
    HCP_Log(INFO, MODULE) << "Enter AfterInitializeTask" << HCPENDLOG;
    m_nasReplicationServiceObj->GetPairID(m_pairID);
    if (!CheckAndCreatePair()) {
        return false;
    }
    // 已有pair有效时检查pair主从关系是否满足要求，如果不满足则需要进行主从切换处理
    SubJobDetails subJobDetails;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetailList;
    ActionResult result;
    ReplicationPairInfo replicationPairInfo;
    if (!m_nasReplicationServiceObj->GetPairInfoByID(m_localDeviceController, m_pairID, replicationPairInfo)) {
        HCP_Log(ERR, MODULE) << "Get pair info fail: " << m_pairID << HCPENDLOG;
        AddLogDetail(logDetail, "plugin_execute_prerequisit_task_fail_label", JobLogLevel::TASK_LOG_ERROR);
        if (IsReportAgentErrorCode(logDetail)) {
            return false;
        }
        AddErrCode(logDetail, HomoErrorCode::REPLICATION_PAIR_NOTEXIST_ERROR_CODE);
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, 0, 0, SubJobStatus::FAILED);
        return false;
    }
    if (!replicationPairInfo.isPrimary) {
        AddLogDetail(logDetail, "", JobLogLevel::TASK_LOG_INFO);
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, 0, 0, SubJobStatus::RUNNING);
        if (!SwitchOverPrimaryAndSecondResource(m_localDeviceController, m_pairID)) {
            HCP_Log(ERR, MODULE) << "SwitchOverPrimaryAndSecondResource fail: " << m_pairID << HCPENDLOG;
            AddLogDetail(logDetail, "plugin_execute_prerequisit_task_fail_label", JobLogLevel::TASK_LOG_ERROR);
            if (IsReportAgentErrorCode(logDetail)) {
                return false;
            }
            AddErrCode(logDetail, m_localDeviceController->GetErrorCode());
            REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, 0, 0, SubJobStatus::FAILED);
            return false;
        }
    }
    HCP_Log(INFO, MODULE) << "Exit AfterInitializeTask" << HCPENDLOG;
    return true;
}

bool ReplicationCommonService::StartTask(std::shared_ptr<DoradoNas>& localController)
{
    HCP_Log(INFO, MODULE) << "Start homo task...." << HCPENDLOG;
    ActionResult result;
    SubJobDetails subJobDetails;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetailList;
    if (!PreStartTask()) {
        HCP_Log(ERR, MODULE) << "PreStartTask fail" << HCPENDLOG;
        return false;
    }
    if (IsAbortJob()) {
        HCP_Log(INFO, MODULE) << "Task is aborted, main job id is " << m_parentJobId << HCPENDLOG;
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, 0, SubJobStatus::ABORTED);
        return false;
    }
    if (!m_nasReplicationServiceObj->OpenPerformanceStatisticSwitch(localController)) {
        HCP_Log(ERR, MODULE) << "OpenPerformanceStatisticSwitch fail" << HCPENDLOG;
        AddLogDetail(logDetail, "dme_replication_synchronize_remote_replication_pair_failed_label",
            JobLogLevel::TASK_LOG_ERROR);
        if (IsReportAgentErrorCode(logDetail)) {
            return false;
        }
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, 0, SubJobStatus::FAILED);
        return false;
    }
    if (!m_nasReplicationServiceObj->SplitReplicationPair(localController, m_pairID)) {
        HCP_Log(WARN, MODULE) << "SplitReplicationPair fail " << m_pairID << HCPENDLOG;
    }
    localController->SetRetryAttr(1);
    m_nasReplicationServiceObj->SecondaryResourceProtectEnable(localController, m_pairID);
    localController->SetRetryAttr(RETRY_TIMES);
    HCP_Log(INFO, MODULE) << "ActiveReplicationPair" << HCPENDLOG;
    std::string productEsn;
    m_localDeviceController->GetESN(productEsn);
    m_productEsn = productEsn;
    if (m_isBackUpFlow) {
        m_satrtTransTime = time(nullptr);
        AddLogDetail(logDetail, "nas_plugin_backup_filesystem_label", JobLogLevel::TASK_LOG_INFO,
            m_productEsn, m_productFSID, m_basicFSID);
    } else {
        AddLogDetail(logDetail, "nas_plugin_restore_filesystem_label", JobLogLevel::TASK_LOG_INFO,
                     m_basicFSID, m_productEsn, m_productFSID);
    }

    REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, 0, SubJobStatus::RUNNING);
    if (!ActiveReplicationPair(localController)) {
        return false;
    }
    HCP_Log(INFO, MODULE) << "Start homo task success" << HCPENDLOG;
    return true;
}

bool ReplicationCommonService::ActiveReplicationPair(std::shared_ptr<DoradoNas>& localController)
{
    ActionResult result;
    SubJobDetails subJobDetails;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetailList;
    int64_t errorCode = 0; // 默认不设置错误码
    if (!m_nasReplicationServiceObj->ActiveReplicationPair(localController, m_pairID, errorCode)) {
        if (m_isBackUpFlow) {
            AddLogDetail(logDetail, "nas_plugin_backup_filesystem_fail_label", JobLogLevel::TASK_LOG_ERROR);
        } else {
            AddLogDetail(logDetail, "nas_plugin_restore_filesystem_fail_label", JobLogLevel::TASK_LOG_ERROR);
        }
        if (IsReportAgentErrorCode(logDetail)) {
            return false;
        }
        AddErrCode(logDetail, errorCode);
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, 0,
                         SubJobStatus::FAILED);
        HCP_Log(ERR, MODULE) << "ActiveReplicationPair fail, error code is " << errorCode << HCPENDLOG;
        return false;
    }
    return true;
}

bool ReplicationCommonService::MonitorExistSnapshotRevert(int64_t &errorCode)
{
    ActionResult result;
    SubJobDetails subJobDetails;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetailList;
    std::string rollbackRate = "";
    std::string rollbackStatus = "0";
    long rollbackRateInt = MAX_ROLLBACK_RATE;
    HCP_Log(INFO, MODULE) << "Wait other snapshot rollback finish" << HCPENDLOG;
    while (true) {
        if (IsAbortJob()) {
            return false;
        }
        sleep(SCAN_TIME_INTERVAL_SECEND);
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, 0, SubJobStatus::RUNNING);
        std::string endtime;
        errorCode = m_nasReplicationServiceObj->QueryRevertEndInfoInStorage(m_localDeviceController,
            rollbackRate, rollbackStatus, endtime, m_snapshotId);
        HCP_Log(INFO, MODULE) << "Query rollbackRate：" << rollbackRate << " rollbackStatus: " << rollbackStatus <<
            " endtime: " << endtime << " snapshotId: " << m_snapshotId << HCPENDLOG;
        if (errorCode == MP_FAILED) {
            HCP_Log(ERR, MODULE) << "Query Revert Info RollbackRate Failed" << HCPENDLOG;
            return false;
        }
        if (rollbackRate != "") {
            rollbackRateInt = std::stol(rollbackRate);
        }
        if ((rollbackStatus == "0") && !endtime.empty()) {
            HCP_Log(INFO, MODULE) << "MonitorExistSnapshotRevert endtime:" << endtime << HCPENDLOG;
            break;
        }
        if ((rollbackStatus == "0") && (rollbackRateInt >= MAX_ROLLBACK_RATE)) {
            HCP_Log(INFO, MODULE) << "MonitorExistSnapshotRevert RollbackRate: " << rollbackRateInt << HCPENDLOG;
            break;
        }
    }
    return true;
}

bool ReplicationCommonService::SwitchOverPrimaryAndSecondResource(std::shared_ptr<DoradoNas>& deviceController,
    const std::string& pairID)
{
    ActionResult result;
    SubJobDetails subJobDetails;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetailList;
    if (pairID == "") {
        HCP_Log(WARN, MODULE) << "PairId is NULL " << HCPENDLOG;
        return true;
    }
    if (!m_nasReplicationServiceObj->SplitReplicationPair(deviceController, pairID)) {
        HCP_Log(ERR, MODULE) << "SplitReplicationPair fail：" << pairID << HCPENDLOG;
        return false;
    }
    if (!m_nasReplicationServiceObj->SecondaryResourceProtectDisable(deviceController, pairID)) {
        HCP_Log(ERR, MODULE) << "Disable secondaryresourceProtect fail"  << HCPENDLOG;
        return false;
    }
    // 取消从端资源保护之后需要monitor状态
    int64_t errorCode = MP_SUCCESS;
    if (!MonitorExistSnapshotRevert(errorCode)) {
        HCP_Log(ERR, MODULE) << "MonitorExistSnapshotRevert fail"  << HCPENDLOG;
        m_nasReplicationServiceObj->SecondaryResourceProtectEnable(deviceController, pairID);
        return false;
    }
    if (IsAbortJob()) {
        HCP_Log(INFO, MODULE) << "Task is aborted, main job id is " << m_parentJobId << HCPENDLOG;
        m_nasReplicationServiceObj->SecondaryResourceProtectEnable(deviceController, pairID);
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, 0, SubJobStatus::ABORTED);
        return false;
    }
    if (!m_nasReplicationServiceObj->SwitchOverPrimaryAndSecondResource(deviceController, pairID)) {
        HCP_Log(ERR, MODULE) << "SwitchOverPrimaryAndSecondResource failed. pairId = " << pairID << HCPENDLOG;
        m_nasReplicationServiceObj->SecondaryResourceProtectEnable(deviceController, pairID);
        return false;
    }
    if (!m_nasReplicationServiceObj->SecondaryResourceProtectEnable(deviceController, pairID)) {
        HCP_Log(ERR, MODULE) << "Enable secondaryresourceProtect fail"  << HCPENDLOG;
        return false;
    }
    HCP_Log(INFO, MODULE) << "SwitchOverPrimaryAndSecondResource success." << HCPENDLOG;
    return true;
}

void ReplicationCommonService::CalculateReplicationProgressAndSpeed(std::shared_ptr<DoradoNas>& localController,
    const uint16_t& previousTaskProgress, const ReplicationPairInfo& replicationPairInfo)
{
    ActionResult result;
    LogDetail logDetail{};
    SubJobDetails subJobDetails;
    std::vector<LogDetail> logDetailList;

    uint32_t taskProgress = 0;
    if (previousTaskProgress <= SYNCHRONIZE_FINISHED) {
        taskProgress = uint32_t((SYNCHRONIZE_FINISHED - previousTaskProgress) *
                         (replicationPairInfo.progress * 1.0 / PROGRESS_OVERALL) + previousTaskProgress);
    }
    m_curProcess = taskProgress;
    uint64_t speed = 0;
    m_nasReplicationServiceObj->GetReplicationSpeed(localController, m_pairID, speed);

    m_speedSamplingTimes++;
    m_totalSpeed += speed;
    m_avgSpeed = m_totalSpeed / m_speedSamplingTimes;
    REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, speed, SubJobStatus::RUNNING);
    return;
}

int64_t ReplicationCommonService::RemotePairMachineProc(std::shared_ptr<DoradoNas>& localController,
    std::shared_ptr<DoradoNas>& remoteController, const ReplicationPairInfo &replicationPairInfo, uint16_t &retryTimes)
{
    ActionResult result;
    SubJobDetails subJobDetails;
    LogDetail logDetail{};
    std::vector<LogDetail> logDetailList;
    SubJobStatus::type status = SubJobStatus::RUNNING;
    int64_t errorCode = 0; // 默认不设置错误码
    if (replicationPairInfo.status == INVALIID) {
        // 无效删除pair, 并返回任务失败
        m_nasReplicationServiceObj->DeleteReplicatonPair(localController, m_pairID);
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, m_avgSpeed, status);
        return HomoErrorCode::REPLICATION_PAIR_INVAILD_ERROR_CODE;
    }
    if (replicationPairInfo.status == TO_BE_RECOVERED) {
        retryTimes = 0;
        HCP_Log(ERR, MODULE) << "Replcation pair is to be recoverd."<< HCPENDLOG;
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, m_avgSpeed, status);
        if (!m_nasReplicationServiceObj->ActiveReplicationPair(localController, m_pairID, errorCode)) {
            HCP_Log(ERR, MODULE) << "ActiveReplicationPair fail, error code is " << errorCode << HCPENDLOG;
            REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, m_avgSpeed, status);
            return errorCode;
        }
    }
    if (replicationPairInfo.status == SPLIT) {
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, m_avgSpeed, status);
        return HomoErrorCode::REPLICATION_PAIR_SPLIT_ERROR_CODE;
    }
    if (!replicationPairInfo.isPrimary) {
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, m_avgSpeed, status);
        return HomoErrorCode::INTERNAL_ERROR_CODE;
    }
    if (replicationPairInfo.status == INTERRUPTED) {
        retryTimes++;
        HCP_Log(DEBUG, MODULE) << "Replcation pair network is INTERRUPTED. Executable times:"
            << retryTimes << HCPENDLOG;
        if (retryTimes == 1) {
            AddLogDetail(logDetail, "nas_plugin_pair_status_interupt_label", JobLogLevel::TASK_LOG_WARNING);
            REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, m_avgSpeed, status);
        } else if (retryTimes >= SCAN_RETRY_TIMES) {
            // 连续1小时链路都是断开状态，任务失败
            REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, m_avgSpeed, status);
            return HomoErrorCode::REPLICATION_PAIR_LINK_TIMEOUT_ERROR_CODE;
        }
    }
    return errorCode;
}

void ReplicationCommonService::DeleteUsedCapacityResource()
{
    /* 恢复不会设置该参数 */
    if (m_capacityPath.empty()) {
        return;
    }
    INFOLOG("Delete capacity path: %s", m_capacityPath.c_str());
    PluginUtils::RemoveFile(m_capacityPath);
}

int64_t ReplicationCommonService::MonitorTask(std::shared_ptr<DoradoNas>& localController,
    std::shared_ptr<DoradoNas>& remoteController)
{
    uint16_t retryTimes = 0;
    uint16_t previousTaskProgress = PROGRESS_TWENTY;
    m_curProcess = previousTaskProgress;
    int64_t errorCode = MP_SUCCESS;
    uint16_t retryPairTimes = 0;
    while (true) {
        ReplicationPairInfo replicationPairInfo;
        if (!m_nasReplicationServiceObj->GetPairInfoByID(localController, m_pairID, replicationPairInfo)) {
            HCP_Log(ERR, MODULE) << "GetPairInfoByID fail, " << m_pairID << HCPENDLOG;
            retryTimes++;
            if (retryTimes >= RETRY_TIMES) {
                HCP_Log(ERR, MODULE) << "Fail to get pair " << m_pairID  << " info over 3 times." << HCPENDLOG;
                return HomoErrorCode::REPLICATION_PAIR_NOTEXIST_ERROR_CODE;
            }
            continue;
        }
        retryTimes = 0;
        if (IsAbortJob()) {
            if (!m_nasReplicationServiceObj->SplitReplicationPair(localController, m_pairID)) {
                HCP_Log(ERR, MODULE) << "SplitReplicationPair fail " << m_pairID << HCPENDLOG;
            }
            return MP_FAILED;
        }
        CalculateReplicationProgressAndSpeed(localController, previousTaskProgress, replicationPairInfo);
        errorCode = RemotePairMachineProc(localController, remoteController, replicationPairInfo, retryPairTimes);
        if (errorCode != MP_SUCCESS) {
            return errorCode;
        }
        if (replicationPairInfo.status == NORMAL) {
            // 任务完成
            return errorCode;
        }
        sleep(SCAN_TIME_INTERVAL_SECEND);
    }
    return errorCode;
}

int64_t ReplicationCommonService::WaitingOtherReplicationPairComplete(std::shared_ptr<DoradoNas>& localController,
    std::shared_ptr<DoradoNas>& remoteController)
{
    HCP_Log(INFO, MODULE) << "Enter WaitingOtherReplicationPairComplete" << HCPENDLOG;
    uint16_t retryTimes = 0;

    while (true) {
        ReplicationPairInfo replicationPairInfo;
        if (!m_nasReplicationServiceObj->GetPairInfoByID(localController, m_pairID, replicationPairInfo)) {
            HCP_Log(ERR, MODULE) << "GetPairInfoByID fail, " << m_pairID << HCPENDLOG;
            retryTimes++;
            if (retryTimes >= RETRY_TIMES) {
                HCP_Log(ERR, MODULE) << "Fail to get pair " << m_pairID  << " info over 3 times." << HCPENDLOG;
                return HomoErrorCode::REPLICATION_PAIR_NOTEXIST_ERROR_CODE;
            }
            continue;
        }
        retryTimes = 0;

        if (replicationPairInfo.status != SYNCHRONIZING) {
            // 等待其他复制pair同步完成
            break;
        }
        sleep(SCAN_TIME_INTERVAL_SECEND);
    }
    return SUCCESS;
}


void ReplicationCommonService::MonitorJobThread(int time)
{
    HCP_Log(INFO, MODULE) << "Enter MonitorJobThread" << HCPENDLOG;
    while (m_connecting) {
        // 60s 上报一次， 1s 等待join
        if (m_reportCount % 60 == 0) {
            ActionResult result;
            SubJobDetails subJobDetails;
            LogDetail logDetail{};
            std::vector<LogDetail> logDetailList;
            HCP_Log(DEBUG, MODULE) << "MonitorJobThread m_jobProgress report : " << m_parentJobId << " , " << HCPENDLOG;
            REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, m_curProcess, 0, SubJobStatus::RUNNING);
        }
        std::this_thread::sleep_for(std::chrono::seconds(time));
        ++m_reportCount;
    }
}

int ReplicationCommonService::GetContainerDoradoIp(std::string& innerDoradoIp)
{
    // 1. if get ip from command line failed, try to get from config map
    innerDoradoIp = k8s::GetConfigMapItem("common-conf", "dorado.mgrip");
    if (!innerDoradoIp.empty()) {
        HCP_Log(DEBUG, MODULE) << "Query dorado management ip address from configmap! IP:"
            << innerDoradoIp << HCPENDLOG;
        return MP_FAILED;
    }

    // 2. get ip from container inner network
    std::vector<std::string> paramList;
    std::vector<std::string> output;
    std::vector<std::string> erroutput;
    std::string cmdSafe("ip route | grep 'via' | grep 'eth0' | grep -v 'default' | awk '{print $3}'");
    int ret = Module::runShellCmdWithOutput(INFO, MODULE, 0, cmdSafe, paramList, output, erroutput);
    if ((ret == 0) && (output.size() == 1)) {
        innerDoradoIp = output[0];
        HCP_Log(INFO, MODULE) << "Query dorado management ip address from ip route! IP:"
            << innerDoradoIp << HCPENDLOG;
        return MP_SUCCESS;
    }
    HCP_Log(ERR, MODULE) << "Get empty dorado ip in the container." << HCPENDLOG;
    return MP_FAILED;
}

bool ReplicationCommonService::IsReportAgentErrorCode(LogDetail logDetail)
{
    ActionResult result;
    SubJobDetails subJobDetails;
    std::vector<LogDetail> logDetailList;
    INFOLOG("IsReportAgentErrorCode m_productManageIps: %s", m_productManageIps.c_str());
    if (PluginUtils::CheckDeviceNetworkConnect(m_productManageIps)) {
        INFOLOG("Network is available");
        return false; // 网络正常场景，无需上报错误码
    }
    HCP_Log(ERR, MODULE) << "Failed to connect the product network" << HCPENDLOG;
    AddErrCode(logDetail, HomoErrorCode::AGENT_SWITCH_ERROR_CODE);
    REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, 0, 0,
        SubJobStatus::FAILED);
    return true;
}

bool ReplicationCommonService::CertVerifyMgr(const ApplicationEnvironment& protectEnv,
    ControlDeviceInfo& productDeviceInfo)
{
    HCP_Log(INFO, MODULE) << "Enter certMgr." << HCPENDLOG;
    CertMgr certMgr;
    if (!certMgr.InitInfo(protectEnv, m_parentJobId)) {
        HCP_Log(ERR, MODULE) << "CertMgr init failed" << HCPENDLOG;
        return false;
    }
    if (!certMgr.IsVerify()) {
        HCP_Log(INFO, MODULE) << "Certificate verification is not required." << HCPENDLOG;
        return true;
    }
    if (!certMgr.FlushCertificationToDisk()) {
        HCP_Log(ERR, MODULE) << "Save cert failed" << HCPENDLOG;
        return false;
    }
    productDeviceInfo.cert = certMgr.GetCertPath();
    if (certMgr.IncludeRevocationList()) {
        if (!certMgr.FlushRevocationListToDisk()) {
            HCP_Log(ERR, MODULE) << "Save CRL failed" << HCPENDLOG;
            return false;
        }
        productDeviceInfo.crl = certMgr.GetCrlPath();
    }
    return true;
}

bool ReplicationCommonService::RemoveCertification(const ApplicationEnvironment& protectEnv)
{
    HCP_Log(INFO, MODULE) << "Enter certMgr." << HCPENDLOG;
    CertMgr certMgr;
    if (!certMgr.InitInfo(protectEnv, m_parentJobId)) {
        HCP_Log(ERR, MODULE) << "CertMgr init failed" << HCPENDLOG;
        return false;
    }
    if (!certMgr.IsVerify()) {
        HCP_Log(INFO, MODULE) << "No certificate needs to be deleted." << HCPENDLOG;
        return true;
    }
    if (!certMgr.RemoveCertAndRevocationListFromDisk()) {
        HCP_Log(ERR, MODULE) << "Remove cert failed" << HCPENDLOG;
        return false;
    }
    return true;
}

void ReplicationCommonService::RemoveSubJobFromChannelMgr()
{
    std::string jobId = GetParentJobId();
    std::string subJobId = GetSubJobId();
    ChannelManager::GetInstance().RemoveSubJob(jobId, subJobId);
}

void ReplicationCommonService::AddSubJobToChannelMgr()
{
    std::string jobId = GetParentJobId();
    std::string subJobId = GetSubJobId();
    ChannelManager::GetInstance().AddSubJob(jobId, subJobId);
}